package com.org.oracle.consumer;

import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.util.concurrent.RateLimiter;
import com.org.oracle.constant.MQConstants;
import com.org.oracle.constant.RedisKeyConstants;
import com.org.oracle.enums.CollectUnCollectNoteTypeEnum;
import com.org.oracle.mapper.NoteCollectionMapper;
import com.org.oracle.mysql.dto.CollectUnCollectNoteMqDTO;
import com.org.oracle.mysql.entity.NoteCollection;
import com.org.oracle.util.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Objects;

/**
 * @author: heiye
 * @date: 2024/12/14 下午8:41
 * @version: v1.0.0
 * @description: 笔记收藏、取消收藏 MQ 消费者
 */
@Slf4j
@Component
@RocketMQMessageListener(
        // Group 组
        consumerGroup = "oracle_group_" + MQConstants.TOPIC_COLLECT_OR_UN_COLLECT,
        // 消费的主题 Topic
        topic = MQConstants.TOPIC_COLLECT_OR_UN_COLLECT
)
public class CollectUnCollectNoteConsumer implements RocketMQListener<Message> {

    // 每秒创建 5000 个令牌
    private RateLimiter rateLimiter = RateLimiter.create(5000);

    @Resource
    private NoteCollectionMapper noteCollectionMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Message message) {
        // 流量削峰：通过获取令牌，如果没有令牌可用，将阻塞，直到获得
        rateLimiter.acquire();

        // 幂等性: 通过联合唯一索引保证

        // 消息体
        String bodyJsonStr = new String(message.getBody());
        // 标签
        String tags = message.getTags();

        log.info("==> CollectUnCollectNoteConsumer 消费了消息 {}, tags: {}", bodyJsonStr, tags);

        // 根据 MQ 标签，判断操作类型
        // 收藏笔记
        if (Objects.equals(tags, MQConstants.TAG_COLLECT)) {
            handleCollectNoteTagMessage(bodyJsonStr);
        }
        // 取消收藏笔记
        else if ((Objects.equals(tags, MQConstants.TAG_UN_COLLECT))) {
            handleUnCollectNoteTagMessage(bodyJsonStr);
        }
    }

    /**
     * 笔记收藏
     *
     * @param bodyJsonStr
     */
    private void handleCollectNoteTagMessage(String bodyJsonStr) {
        // 消息体 JSON 字符串转 DTO
        CollectUnCollectNoteMqDTO collectNoteMqDTO = JsonUtils.parseObject(bodyJsonStr, CollectUnCollectNoteMqDTO.class);

        if (Objects.isNull(collectNoteMqDTO)) {
            return;
        }

        // 用户ID
        String uid = collectNoteMqDTO.getUid();
        // 收藏的笔记ID
        String noteId = collectNoteMqDTO.getNoteId();
        // 操作类型
        Integer type = collectNoteMqDTO.getType();
        // 收藏时间
        LocalDateTime createTime = collectNoteMqDTO.getCreateTime();

        // 构建对象
        NoteCollection noteCollection = NoteCollection.builder()
                .uid(uid)
                .noteId(noteId)
                .status(type)
                .createTime(createTime)
                .build();

        // 添加或更新笔记收藏记录
        int count = noteCollectionMapper.saveOrUpdate(noteCollection);

        if (count == 0) {
            return;
        }

        // 更新数据库成功后，发送计数 MQ
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyJsonStr).build();

        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_COLLECT, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数: 笔记收藏】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数: 笔记收藏】MQ 发送异常: ", throwable);
            }
        });
    }

    /**
     * 笔记取消收藏
     *
     * @param bodyJsonStr
     */
    private void handleUnCollectNoteTagMessage(String bodyJsonStr) {
        // 消息体 JSON 字符串转 DTO
        CollectUnCollectNoteMqDTO unCollectNoteMqDTO = JsonUtils.parseObject(bodyJsonStr, CollectUnCollectNoteMqDTO.class);

        if (Objects.isNull(unCollectNoteMqDTO)) {
            return;
        }

        // 构建更改对象
        LambdaUpdateWrapper<NoteCollection> lambdaUpdateWrapper = Wrappers.<NoteCollection>lambdaUpdate()
                // 操作类型
                .set(NoteCollection::getStatus, unCollectNoteMqDTO.getType())
                // 创建时间
                .set(NoteCollection::getCreateTime, unCollectNoteMqDTO.getCreateTime())
                // 用户ID
                .eq(NoteCollection::getUid, unCollectNoteMqDTO.getUid())
                // 笔记ID
                .eq(NoteCollection::getNoteId, unCollectNoteMqDTO.getNoteId())
                // 确保布隆过滤器误判的情况下也能更改成功
                .eq(NoteCollection::getStatus, CollectUnCollectNoteTypeEnum.COLLECT.getCode());

        // 取消收藏：记录更新
        int count = noteCollectionMapper.update(lambdaUpdateWrapper);

        if (count == 0) {
            return;
        }

        // 更新数据库成功后，发送计数 MQ
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(bodyJsonStr).build();

        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_COLLECT, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数: 笔记取消收藏】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数: 笔记取消收藏】MQ 发送异常: ", throwable);
            }
        });
    }
}
